﻿using System;
using System.Collections.Generic;
using System.Threading;
using PSE.Framework.Threading;

namespace PSE.Framework.Messaging
{
    /// <summary>
    /// Represents a first-in, first-out synchronized(thread-safe) collection of objects.
    /// </summary>
    /// <typeparam name="T">Specifies the type of elements in the queue.</typeparam>
    [Serializable]
    public class SynchronizedQueue<T> : IDisposable
    {
        #region Constants
        const int TIME_TO_CHECK_MESSAGES_QUEUED = 93;
        const int TIME_TO_HEALTHSTATUS_CHECK = 2000;
        #endregion

        #region Delegates
        /// <summary>
        /// Represents the method that will receive the messages one by one.
        /// </summary>
        /// <param name="sender">The object itself.</param>
        /// <param name="item">The message to be received.</param>
        /// <param name="queueId">The queue's id.</param>
        public delegate void SynchronizedQueueMessageHandler(SynchronizedQueue<T> sender, T item, string queueId);
        /// <summary>
        /// Represents the method that will receive the messages per block.
        /// </summary>
        /// <param name="sender">The object itself.</param>
        /// <param name="item">The list of messages to be received.</param>
        /// <param name="queueId">The queue's id.</param>
        public delegate void SynchronizedQueueMessageListHandler(SynchronizedQueue<T> sender, List<T> item, string queueId);
        /// <summary>
        /// Represents the method that will handle a health status change.
        /// </summary>
        /// <param name="queueId">The queue's id.</param>
        /// <param name="synchronizedQueue">The object itself.</param>
        public delegate void HealthStatusChangedHandler(string queueId, SynchronizedQueue<T> synchronizedQueue);
        #endregion

        #region Attributes
        private Queue<T> _queue;
        private DateTime _lastHealthStatusChecked = DateTime.Now;
        private HealthStatusChangedHandler _healthStatusChangedHandler;
        private bool _keepRunning = false;
        private bool _keepMonitoring = false;
        private bool _paused = false;
        private bool _messageArrivalSubscribed;
        private bool _messageListArrivalSubscribed;
        private string _queueId;
        private int _totalMessageSentPerInterval = 0;
        private int _messagesSentPerSecond;
        private int _totalMessageReceivedPerInterval = 0;
        private int _messagesReceivedPerSecond = 0;
        private SynchronizedQueue<T>.SynchronizedQueueMessageHandler _messageArrival;
        private SynchronizedQueue<T>.SynchronizedQueueMessageListHandler _messageArrivalList;
        private int _maxItemsBlock = int.MaxValue;
        private MessageArrivalType _messageArrivalType;
        #endregion

        #region Constructors
        /// <summary>
        /// Initializes a new instance of the PSE.Framework.Messaging.SynchronizedQueue class.
        /// </summary>
        /// <param name="queueId">The queue's id.</param>
        /// <param name="healthStatusChangedHandler">The method that will handle a health status change.</param>
        public SynchronizedQueue(string queueId, HealthStatusChangedHandler healthStatusChangedHandler)
        {
            if (string.IsNullOrEmpty(queueId))
                throw new ArgumentNullException("queueId");

            if (healthStatusChangedHandler == null)
                throw new ArgumentNullException("healthStatusChangedHandler");

            _queue = new Queue<T>();
            _queueId = queueId;
            _healthStatusChangedHandler = healthStatusChangedHandler;
            StartMonitor();
        }
        /// <summary>
        /// Initializes a new instance of the PSE.Framework.Messaging.SynchronizedQueue class.
        /// </summary>
        /// <param name="queueId">The queue's id.</param>
        /// <param name="queue">The queue</param>
        /// <param name="healthStatusChangedHandler">The method that will handle a health status change.</param>
        public SynchronizedQueue(string queueId, Queue<T> queue, HealthStatusChangedHandler healthStatusChangedHandler)
        {
            if (string.IsNullOrEmpty(queueId))
                throw new ArgumentNullException("queueId");

            if (queue == null)
                throw new ArgumentNullException("queue");

            if (healthStatusChangedHandler == null)
                throw new ArgumentNullException("healthStatusChangedHandler");

            _queue = queue;
            _queueId = queueId;
            _healthStatusChangedHandler = healthStatusChangedHandler;

            StartMonitor();
        }
        #endregion

        #region Properties
        /// <summary>
        /// Gets or sets the maximum number of messages to be dequeued per block.
        /// </summary>
        public int MaxItemsBlock
        {
            get { return _maxItemsBlock; }
            set { _maxItemsBlock = value; }
        }
        /// <summary>
        /// Gets or sets the message's delivery type.
        /// </summary>
        public MessageArrivalType MessageArrivalType
        {
            get { return _messageArrivalType; }
            set
            {
                _messageArrivalType = value;
                if (_messageArrivalType == MessageArrivalType.NonBlock)
                {
                    MaxItemsBlock = 1;
                }
                else if (_messageArrivalType == MessageArrivalType.Block)
                {
                    if (MaxItemsBlock <= 0)
                    {
                        MaxItemsBlock = int.MaxValue;
                    }
                }
            }
        }
        /// <summary>
        /// Gets a value indicating whether MessageListArrival handler is subscribed.
        /// </summary>
        public bool MessageListArrivalSubscribed
        {
            get { return _messageListArrivalSubscribed; }
        }
        /// <summary>
        /// Gets a value indicating whether MessageArrival handler is subscribed.
        /// </summary>
        public bool MessageArrivalSubscribed
        {
            get { return _messageArrivalSubscribed; }
        }
        /// <summary>
        /// Gets or sets the method that will receive the messages one by one.
        /// </summary>
        public SynchronizedQueue<T>.SynchronizedQueueMessageHandler MessageArrival
        {
            get { return _messageArrival; }
            set
            {
                _messageArrival = value;
                _messageArrivalSubscribed = true;
                StartReader();
            }
        }
        /// <summary>
        /// Gets or sets the method that will receive the messages per block.
        /// </summary>
        public SynchronizedQueue<T>.SynchronizedQueueMessageListHandler MessageArrivalList
        {
            get { return _messageArrivalList; }
            set
            {
                _messageArrivalList = value;
                _messageListArrivalSubscribed = true;
                StartReader();
            }
        }
        /// <summary>
        /// Gets the number of messages sent per second.
        /// </summary>
        public int SentMessagesPerSecond
        {
            get { return _messagesSentPerSecond; }
        }
        /// <summary>
        /// Gets the number of messages received per second.
        /// </summary>
        public int ReceivedMessagesPerSecond
        {
            get { return _messagesReceivedPerSecond; }
        }

        #endregion

        #region Methods
        #region Métodos para o Receiver
        private void StartReader()
        {
            if (!_keepRunning)
            {
                Thread thr1 = new Thread(new ThreadStart(Run));
                thr1.Name = "Internal Queue Manager - " + _queueId;
                thr1.Start();
            }
        }

        private void Run()
        {
            _keepRunning = true;

            while (_keepRunning)
            {
                if (_queue.Count > 0 && (!_paused))
                {
                    if (_messageArrivalType == MessageArrivalType.Block)
                    {
                        DequeueInBlocks();
                    }
                    else if (_messageArrivalType == MessageArrivalType.NonBlock)
                    {
                        DequeueOneByOne();
                    }
                }

                if (_queue.Count <= 0)
                    Sleeper.Sleep(TIME_TO_CHECK_MESSAGES_QUEUED, IsRunning);
            }
        }

        private void DequeueInBlocks()
        {
            List<T> listItens = new List<T>();
            int cont;
            int itensProc;

            lock (_queue)
            {
                cont = _queue.Count;
            }

            itensProc = 0;
            while (cont > 0 && itensProc < MaxItemsBlock && _keepRunning && (!_paused))
            {
                lock (_queue)
                    listItens.Add(_queue.Dequeue());

                itensProc++;
                cont--;
            }

            _messageArrivalList(this, listItens, _queueId);
            _totalMessageReceivedPerInterval += listItens.Count;
        }

        private void DequeueOneByOne()
        {
            int cont = 0;

            lock (_queue)
            {
                cont = _queue.Count;
            }
            while (cont > 0 && _keepRunning && (!_paused))
            {
                T toArrive = default(T);

                lock (_queue)
                {
                    if (_queue.Count > 0)
                        toArrive = _queue.Dequeue();

                    cont--;
                }
                _messageArrival(this, toArrive, _queueId);
                _totalMessageReceivedPerInterval++;
            }
        }

        private bool IsRunning()
        {
            return _keepRunning;
        }

        private bool IsMonitoring()
        {
            return _keepMonitoring;
        }

        /// <summary>
        /// Pauses to dequeue messages.
        /// </summary>
        public void Pause()
        {
            _paused = true;
        }

        /// <summary>
        /// Resumes to dequeue messages.
        /// </summary>
        public void Resume()
        {
            _paused = false;
        }
        #endregion

        #region Métodos de Manipulação da Fila
        /// <summary>
        /// Adds an object to the end of the System.Collections.Generic.Queue&lt;T&gt;.
        /// </summary>
        /// <param name="item">The object to add to the System.Collections.Generic.Queue&lt;T&gt;. The value can be null for reference types.</param>
        public void Enqueue(T item)
        {
            if (item == null)
                throw new ArgumentNullException("item");

            lock (_queue)
                _queue.Enqueue(item);

            _totalMessageSentPerInterval++;
        }
        /// <summary>
        /// Removes and returns the object at the beginning of the System.Collections.Generic.Queue&lt;T&gt;.
        /// </summary>
        /// <returns>The object that is removed from the beginning of the System.Collections.Generic.Queue&lt;T&gt;.</returns>
        public T Dequeue()
        {
            T ret;

            lock (_queue)
                ret = _queue.Dequeue();

            _totalMessageReceivedPerInterval++;

            return ret;
        }

        /// <summary>
        /// Gets the number of elements contained in the System.Collections.Generic.Queue&lt;T&gt;.
        /// </summary>        
        public int Count
        {
            get
            {
                lock (_queue)
                    return _queue.Count;
            }
        }
        /// <summary>
        /// Returns the object at the beginning of the System.Collections.Generic.Queue&lt;T&gt; without removing it.
        /// </summary>
        /// <returns>The object at the beginning of the System.Collections.Generic.Queue&lt;T&gt;.</returns>        
        public T Peek()
        {
            T ret;

            lock (_queue)
                ret = _queue.Peek();

            return ret;
        }
        /// <summary>
        /// Removes all objects from the System.Collections.Generic.Queue&lt;T&gt;.
        /// </summary>
        public void Clear()
        {
            lock (_queue)
                _queue.Clear();
        }
        #endregion

        #region Métodos de monitoração da fila
        /// <summary>
        /// 
        /// </summary>
        private void StartMonitor()
        {
            Thread thr1 = new Thread(new ThreadStart(MonitorQueue));
            thr1.Name = "Queue Monitor - " + _queueId;
            thr1.Start();
        }
        /// <summary>
        /// 
        /// </summary>
        private void MonitorQueue()
        {
            _keepMonitoring = true;

            while (_keepMonitoring)
            {
                CalculateMessageNumberPerSecond();

                if (_healthStatusChangedHandler != null)
                    _healthStatusChangedHandler(_queueId, this);

                Sleeper.Sleep(TIME_TO_HEALTHSTATUS_CHECK, IsMonitoring);
            }
        }
        /// <summary>
        /// 
        /// </summary>
        private void CalculateMessageNumberPerSecond()
        {
            int seconds;

            // Calcula o número de segundos entre a última coleta e a coleta atual
            seconds = DateTime.Now.Subtract(_lastHealthStatusChecked).Seconds;

            if (seconds == 0)
                seconds = TIME_TO_HEALTHSTATUS_CHECK / 1000;

            if (_totalMessageSentPerInterval >= seconds)
                _messagesSentPerSecond = _totalMessageSentPerInterval / seconds;
            else
                _messagesSentPerSecond = _totalMessageSentPerInterval;

            if (_messagesReceivedPerSecond >= seconds)
                _messagesReceivedPerSecond = _totalMessageReceivedPerInterval / seconds;
            else
                _messagesReceivedPerSecond = _totalMessageReceivedPerInterval;

            // Zera o contador que será usado na próxima coleta
            // Esse contador é incrementado dentro do Send(Message message)
            // a cada mensagem enviada
            _totalMessageSentPerInterval = 0;
            _totalMessageReceivedPerInterval = 0;

            // Define o horário dessa verificação, para ser usada na próxima chamada da função e calcular os segundos entre uma chamada e outra
            _lastHealthStatusChecked = DateTime.Now;

        }
        /// <summary>
        /// 
        /// </summary>
        /// <param name="healthStatusChangedHandler"></param>
        private void SetHealthStatusChangedHandler(HealthStatusChangedHandler healthStatusChangedHandler)
        {
            if (healthStatusChangedHandler == null)
                throw new ArgumentNullException("healthStatusChangedHandler");

            _healthStatusChangedHandler = healthStatusChangedHandler;
        }
        #endregion

        #region IDisposable Members
        /// <summary>
        /// Releases all resources.
        /// </summary>
        public void Dispose()
        {
            _keepRunning = false;
            _keepMonitoring = false;
        }
        #endregion
        #endregion
    }
}
